/*
 * Copyright (c) 2006-2021, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2022-05-26     Administrator       the first version
 */
#include "mqtt_protocol.h"
#include <rtthread.h>
#define DBG_ENABLE
#define DBG_SECTION_NAME    "mqtt.interface"
#define DBG_LEVEL           DBG_LOG
#define DBG_COLOR
#include <rtdbg.h>
#include "paho_mqtt.h"


#define CLIENTID "00000001"
#define topic_up "/topic/up/00000001"
#define topic_down "/topic/down/00000001"
#define user_name "test1"
#define pass_word "123456"

unsigned char serverUrl[256];

//#define MQTT_URI                "tcp://10.10.10.23:1883"
char MQTT_URI[256];
#define MQTT_SUBTOPIC           topic_down
#define MQTT_PUBTOPIC           topic_up
#define MQTT_WILLMSG            "Goodbye!"

/* define MQTT client context */
static MQTTClient client;
static int is_started = 0;

void initMqttDefaultParam(void)
{
    memcpy(MQTT_URI,"tcp://192.168.0.38:1883",strlen("tcp://192.168.0.38:1883"));
}

void reInitMqttServerInfo(unsigned char *serverIp,unsigned short portNum)
{
    sprintf(MQTT_URI,"tcp://%s:%d",serverIp,portNum);
}

int paho_mqtt_publish_direct(MQTTClient *client, enum QoS qos, const char *topic, const char *msg_str)
{
    MQTTMessage message;

    if (qos != QOS1)
    {
        LOG_E("Not support Qos(%d) config, only support Qos(d).", qos, QOS1);
        return PAHO_FAILURE;
    }

    message.qos = qos;
    message.retained = 0;
    message.payload = (void *)msg_str;
    message.payloadlen = rt_strlen(message.payload);
    return mqtt_send_buffer(client,&message,topic);
}

int mqtt_interface_send(unsigned char *pBuf,unsigned short count)
{
    int ret;
    if (is_started == 0)
    {
        //LOG_E("mqtt client is not connected.");
        return -1;
    }
    pBuf[count] = 0;
    ret = paho_mqtt_publish_direct(&client, QOS1, MQTT_PUBTOPIC, pBuf);
    if(ret!=0){
        //LOG_E("mqtt_interface_send ERROR!");
    }

    return ret;
}
static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}

static void mqtt_sub_default_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt sub default callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}

static void mqtt_connect_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_connect_callback!");
}

static void mqtt_online_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_online_callback!");
}

static void mqtt_offline_callback(MQTTClient *c)
{
    LOG_D("inter mqtt_offline_callback!");
}

int mqtt_start(void)
{
    /* init condata param by using MQTTPacket_connectData_initializer */
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
    static char cid[20] = { 0 };

    if (is_started)
    {
        LOG_E("mqtt client is already connected.");
        return -1;
    }

    /* config MQTT context param */
    {
        client.isconnected = 0;
        client.uri = MQTT_URI;

        /* generate the random client ID */
        rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());
        /* config connect param */
        rt_memcpy(&client.condata, &condata, sizeof(condata));
        client.condata.clientID.cstring = CLIENTID;
        client.condata.keepAliveInterval = 30;
        client.condata.cleansession = 1;
        client.condata.username.cstring = user_name;
        client.condata.password.cstring = pass_word;
        /* config MQTT will param. */
        client.condata.willFlag = 1;
        client.condata.will.qos = 1;
        client.condata.will.retained = 0;
        client.condata.will.topicName.cstring = MQTT_PUBTOPIC;
        client.condata.will.message.cstring = MQTT_WILLMSG;

        /* rt_malloc buffer. */
        client.buf_size = client.readbuf_size = 1024;
        client.buf = rt_calloc(1, client.buf_size);
        client.readbuf = rt_calloc(1, client.readbuf_size);
        if (!(client.buf && client.readbuf))
        {
            LOG_E("no memory for MQTT client buffer!");
            return -1;
        }

        /* set event callback function */
        client.connect_callback = mqtt_connect_callback;
        client.online_callback = mqtt_online_callback;
        client.offline_callback = mqtt_offline_callback;

        /* set subscribe table and event callback */
        client.messageHandlers[0].topicFilter = rt_strdup(MQTT_SUBTOPIC);
        client.messageHandlers[0].callback = mqtt_sub_callback;
        client.messageHandlers[0].qos = QOS1;

        /* set default subscribe event callback */
        client.defaultMessageHandler = mqtt_sub_default_callback;
    }

    /* run mqtt client */
    paho_mqtt_start(&client);
    is_started = 1;

    return 0;
}


int mqtt_stop(void)
{
    is_started = 0;

    return paho_mqtt_stop(&client);
}

int mqtt_publish(unsigned char *buffer, unsigned short len)
{
    if (is_started == 0)
    {
        LOG_E("mqtt client is not connected.");
        return -1;
    }

    return paho_mqtt_publish(&client, QOS1, MQTT_PUBTOPIC, buffer);
}

static void mqtt_new_sub_callback(MQTTClient *client, MessageData *msg_data)
{
    *((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';
    LOG_D("mqtt new subscribe callback: %.*s %.*s",
               msg_data->topicName->lenstring.len,
               msg_data->topicName->lenstring.data,
               msg_data->message->payloadlen,
               (char *)msg_data->message->payload);
}
#ifdef FINSH_USING_MSH

int mqtt_config(int argc, char **argv)
{
    if (argc == 1)
    {
        rt_kprintf("bad parameter! e.g: mqtt_config 192.168.0.0 8802\n");
    }
    else if (argc == 3)
    {
        rt_kprintf("serverIp : %s\n", argv[1]);
        rt_kprintf("serverPort: %d\n", atoi(argv[2]));
        reInitMqttServerInfo(argv[1],atoi(argv[2]));
        mqtt_stop();
        sleep(5);
        mqtt_start();
    }
    else
    {
        rt_kprintf("bad parameter! e.g: mqtt_config 192.168.0.0 8802\n");
    }

    return 0;
}
MSH_CMD_EXPORT(mqtt_config, configure server ip and port);
#endif
/*
int init_mqttClientWithServerIp(unsigned char *ipServer,unsigned short portNum,char *userName,char *password)
{
    int ret;
    memset(serverUrl,0,256);
    sprintf(serverUrl,"tcp://%s:%d",ipServer,portNum);

    ret = init_mqtt_interface(serverUrl,CLIENTID,
            userName,password,topic_up);
    return 0;
}
*/
